package com.xwj.comsumer;

import com.xwj.service.EvenQueueService;
import com.xwj.service.OddQueueService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * @author xwj
 * @date 2020/3/26
 */
@Component
public class KafkaConsumer {
    private static Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
    @Autowired
    private OddQueueService oddQueueService;
    @Autowired
    private EvenQueueService evenQueueService;

//    static {
//        new Thread(new OddQueueConsumer()).start();
//        new Thread(new EvenQueueConsumer()).start();
//    }
//
//    /**
//     * 奇数key队列
//     */
//    private static final LinkedBlockingQueue<Object> ODD_QUEUE = new LinkedBlockingQueue<Object>(100);
//
//    static class OddQueueConsumer implements Runnable {
//        @Override
//        public void run() {
//            while (true) {
//                Object value = null;
//                try {
//                    value = ODD_QUEUE.take();
//                    System.out.println("奇数内存队列消费：" + value);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }
//        }
//    }
//
//    /**
//     * 偶数key队列
//     */
//    private static final LinkedBlockingQueue<Object> EVEN_QUEUE = new LinkedBlockingQueue<Object>(100);
//
//    static class EvenQueueConsumer implements Runnable {
//        @Override
//        public void run() {
//            while (true) {
//                Object value = null;
//                try {
//                    value = EVEN_QUEUE.take();
//                    System.out.println("偶数内存队列消费：" + value);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }
//        }
//    }


    /**
     * group1消费组：消费者1
     *
     * @param consumerRecord
     * @param ack
     */
    @KafkaListener(topics = {"xwj-topic-1"}, groupId = "group1")
    public void consumer1_1(ConsumerRecord<String, Object> consumerRecord, Acknowledgment ack) {
        logger.info("group1消费组：消费者1开始消费");
        consumer1(consumerRecord, ack);
    }


    /*===========同一消费组多个消费者消费同一主题中的消息============*/
    /**
     * group1消费组：消费者2
     *
     * @param consumerRecord
     * @param ack
     */
    @KafkaListener(topics = {"xwj-topic-1"}, groupId = "group1")
    public void consumer1_2(ConsumerRecord<String, Object> consumerRecord, Acknowledgment ack) {
        logger.info("group1消费组：消费者2开始消费");
        consumer1(consumerRecord, ack);
    }
    /*===========同一消费组多个消费者消费同一主题中的消息============*/


    /*===========两个消费组同时消费统一主题中的消息============*/
    /**
     * 两个消费组消费统一主题中的消息，消费组group1和消费组group3都能收到xwj-topic-1主题的全量消息，在逻辑意义上来说它们属于不同的应用
     *
     * group1消费组：消费者2
     *
     * @param consumerRecord
     * @param ack
     */
//    @KafkaListener(topics = {"xwj-topic-1"}, groupId = "group3")
//    public void consumer3(ConsumerRecord<String, Object> consumerRecord, Acknowledgment ack) {
//        logger.info("group3消费组：消费者开始消费");
//        consumer1(consumerRecord, ack);
//    }
    /*===========两个消费组同时消费统一主题中的消息============*/

    private void consumer1(ConsumerRecord<String, Object> consumerRecord, Acknowledgment ack) {
        // 根据key路由到不同的队列中进行消费，每个线程消费一个队列，多线程处理提高吞吐量
        // 此处以奇数、偶数key作为区分，奇数key进入奇数队列中，偶数key进入偶数队列中
        if (!StringUtils.isEmpty(consumerRecord.key())) {
            if ("odd-key".equals(consumerRecord.key())) {
//                ODD_QUEUE.offer(consumerRecord.value());
                oddQueueService.consumer(consumerRecord.value());
            } else {
//                EVEN_QUEUE.offer(consumerRecord.value());
                evenQueueService.consumer(consumerRecord.value());
            }
        }
        logger.info("消费者消费topic:{},key:{},partition:{}的消息：{}", consumerRecord.topic(), consumerRecord.key(), consumerRecord.partition(),
                consumerRecord.value());
        // 手动提交
        ack.acknowledge();
    }

    @KafkaListener(topics = {"xwj-topic-2"}, groupId = "group2")
    public void consumer2(ConsumerRecord<String, Object> consumerRecord) {
        logger.info("消费者消费topic:{},partition:{}的消息：{}", consumerRecord.topic(), consumerRecord.partition(),
                consumerRecord.value());
    }
}
